package com.qingcloud.df.executor.cluster;

import com.qingcloud.df.sdk.component.constains.Constants;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.message.*;
import com.qingcloud.df.sdk.component.message.request.InstanceStartRequest;
import com.qingcloud.df.sdk.component.message.request.JobRequest;
import com.qingcloud.df.sdk.component.message.response.AkkaResponse;
import com.qingcloud.df.sdk.component.message.response.JobResponse;
import com.qingcloud.df.sdk.component.message.response.NodeContextReponse;
import com.qingcloud.df.sdk.component.transport.actor.AkkaActor;
import com.qingcloud.df.sdk.component.transport.actor.AkkaHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * 集群主节点，接收心跳，节点注册等数据
 */

@Service
@AkkaActor(path = Constants.E4E_PATH)
public class ServerActorHandler {

    Logger logger = LoggerFactory.getLogger(ServerActorHandler.class);

    @Resource
    private ServerService serverService;
    /**
     * 执行节点心跳数据 Executor-->Server
     * @param reportHealth
     */
    @AkkaHandler(path = Constants.ES_HANDLER_HEARTBEAT)
    public AkkaResponse<Pong> executorHeartbeat(ReportHealth reportHealth) {
        logger.debug("---->receive node[{}] heartbeat<--------------------",reportHealth.getAkkaAddress());
        serverService.nodeHeartbeatCheck(reportHealth.getNodeInfo());
        Pong pong = new Pong();
        pong.setKnownNode(true);
        return AkkaResponse.success(pong);
    }

    @AkkaHandler(path = Constants.ES_HANDLER_EXECUTOR_PINT)
    public AkkaResponse<Pong>  pong(Ping ping) {
        logger.debug("------------------->receive executor ping request<--------------------");
        Pong pong = serverService.executorPing(ping);
        return AkkaResponse.success(pong);
    }

    /**
     * 执行节点注册   Executor-->Server
     * @param nodeinfo
     */
    @AkkaHandler(path = Constants.ES_HANDLER_REGISTER)
    public AkkaResponse<String> executorRegister(NodeInfo nodeinfo) {
        logger.debug("---->receive node[{}] register request<--------------------",nodeinfo.getAkkaAddress());
        serverService.nodeRegister(nodeinfo);
        return AkkaResponse.success("registered");
    }

    @AkkaHandler(path = Constants.ES_HANDLER_PULLJOB)
    public AkkaResponse<List<InstanceStartRequest>> executorPullJob(JobRequest request)  {
        logger.debug("---->receive node[{}] pull job request <--------------------",request.getNodeId());
        List<InstanceStartRequest> requests = serverService.buildJobConfig(request.getNodeId(),request.getJobList());
        return AkkaResponse.success(requests);
    }

    @AkkaHandler(path = Constants.ES_HANDLER_JOB_CONTEXT)
    public void updataJobContext(JobResponse response) {
        String nodeId = ClusterManager.getNodeByJobId(response.getJobId());
        logger.debug("---->job[{}] as assign to node[{}]  successful！ <--------------------",response.getJobId());
        ClusterManager.pubNodeJob(nodeId,response.getJobId(),response.getContextMap());
    }

    @AkkaHandler(path = Constants.ES_HANDLER_NODE_CONTEXT)
    public void updataNodeContext(NodeContextReponse nodeContextReponse) {
        int jobSize = 0 ;
        if(nodeContextReponse.getContextMap()!=null) {
            jobSize = nodeContextReponse.getContextMap().size();
        }
        logger.debug("---->receive node[{}] update info,  all[{}] jobs has reassign it  <--------------------",nodeContextReponse.getNodeId(),jobSize);
        ClusterManager.pubNode(nodeContextReponse.getNodeId(),nodeContextReponse.getContextMap());
    }
}
